package com.ecarx.library_base.scoket;

import android.text.TextUtils;
import android.util.Log;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineException;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author WL on 2023/04/18
 */
final class SocketThread extends Thread {
    private static final String TAG = SocketThread.class.getName();
    // 服务 Host
    private final String mHost;
    // 服务端口
    private final int mPort;

    private ChannelFactory mChannelFactory;
    private ClientBootstrap mClientBootstrap;
    private volatile ChannelFuture mChannelFuture;
    private volatile Channel mChannel;
    // 超时间的增长因子，步进为 STEP_TIMEOUT
    private final AtomicInteger FACTOR = new AtomicInteger(1);
    private final int MAX_FACTOR = 24;
    // 超时时间的步长为 5 秒
    private final int STEP_TIMEOUT = 5000;
    // 是否继续标志
    private final AtomicBoolean CONTINUE_FLAG = new AtomicBoolean(Boolean.TRUE);
    private SimpleChannelHandler mHanlder;

    public SocketThread(String host, int port, SimpleChannelHandler handler) {
        this.mHost = host;
        this.mPort = port;
        this.mHanlder = handler;
        initChannel(handler);
    }

    private void initChannel(final SimpleChannelHandler channelHandler) {
        // only one IO thread
        mChannelFactory = new NioClientSocketChannelFactory(
                Executors.newSingleThreadExecutor(),
                Executors.newSingleThreadExecutor());

        mClientBootstrap = new ClientBootstrap(mChannelFactory);

        mClientBootstrap.setPipelineFactory(() -> {
            // 1.
            ChannelPipeline pipeline = Channels.pipeline();
            // 2. 接收的数据包解码
//            pipeline.addLast("decoder", new LengthFieldBasedFrameDecoder(
//                    400 * 1024, 0, 4,
//                    -4, 0));
            // 3. 发送的数据包编码
            //pipeline.addLast("encoder", new PacketEncoder());
            // 4. 具体的业务处理，这个handler只负责接收数据，并传递给dispatcher
            pipeline.addLast("handler", channelHandler);
            return pipeline;

        });

        mClientBootstrap.setOption("tcpNoDelay", true);
        mClientBootstrap.setOption("keepAlive", true);
    }

    @Override
    public void run() {
        super.run();
        while (FACTOR.get() <= MAX_FACTOR && CONTINUE_FLAG.get()) {
            // 连接超时策略：最长的连接时间为 2 分钟，初始的连接超时时间为5秒
            mClientBootstrap.setOption("connectTimeoutMillis", FACTOR.get() * STEP_TIMEOUT);
            try {
                boolean connect = doConnect(Boolean.FALSE);
                if (!connect) {
                    initChannel(mHanlder);
                }
            } catch (ClassCastException cce) {
                Log.e(TAG, String.format("%s:%s -- ClassCastException \n %s", mHost, mHost, cce.getMessage()));
            } catch (ChannelPipelineException cpe) {
                Log.e(TAG, String.format("%s:%s -- ChannelPipelineException \n %s", mHost, mHost, cpe.getMessage()));
            } catch (RejectedExecutionException ree) {
                Log.e(TAG, String.format("%s:%s -- RejectedExecutionException \n %s", mHost, mHost, ree.getMessage()));
                initChannel(mHanlder);
            } catch (RuntimeException re) {
                Log.e(TAG, String.format("%s:%s -- RuntimeException \n %s", mHost, mHost, re.getMessage()));
            } catch (Exception e) {
                Log.e(TAG, String.format("%s:%s -- Exception \n %s", mHost, mHost, e.getMessage()));
            } finally {
                FACTOR.getAndIncrement();
            }
        }

        if (mClientBootstrap != null) {
            // Shut down thread pools to exit.
            try {
                mClientBootstrap.releaseExternalResources();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public final boolean doConnect(boolean reset) {
        // Host 不能为空
        if (TextUtils.isEmpty(mHost)) {
            return Boolean.FALSE;
        }

        // 端口号不能为空
        if (mPort <= 0) {
            return Boolean.FALSE;
        }

        // 如果是已连接的服务，则判断连接的服务是否目标服务，如果是则直接返回 True，否则继续执行
        if (mChannel != null && mChannel.isConnected()) {
            return Boolean.TRUE;
        }

        // Socket 的目标地址
        final InetSocketAddress destAddress = new InetSocketAddress(mHost, mPort);

        mChannelFuture = mClientBootstrap.connect(destAddress);

        // Wait until the connection attempt succeeds or fails.
        mChannel = mChannelFuture.awaitUninterruptibly().getChannel();
        if (!mChannelFuture.isSuccess()) {
            mChannelFuture.getCause().printStackTrace();
            mClientBootstrap.releaseExternalResources();
            try {
                Thread.sleep(FACTOR.get() * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return Boolean.FALSE;
        }
        FACTOR.set(0);

        // Wait until the connection is closed or the connection attemp
        // fails.
        mChannelFuture.getChannel().getCloseFuture().awaitUninterruptibly();
        try {
            Thread.sleep(FACTOR.get() * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }

    public boolean close() {
        CONTINUE_FLAG.set(Boolean.FALSE);
        if (null == mChannelFuture)
            return Boolean.TRUE;
        try {
            if (null != mChannelFuture.getChannel() && mChannelFuture.getChannel().isConnected()) {
                mChannelFuture.getChannel().close();
            }
            mChannelFuture.cancel();
            return Boolean.TRUE;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }

    public boolean isConnected() {
        if (mChannelFuture == null) {
            return Boolean.FALSE;
        }

        Channel channel = mChannelFuture.getChannel();
        if (channel == null) {
            return Boolean.FALSE;
        }

        return channel.isConnected();
    }

    public boolean sendRequest(String requset) {
        if (TextUtils.isEmpty(requset)) {
            return Boolean.FALSE;
        }

        Channel currentChannel = mChannelFuture.getChannel();
        if (currentChannel != null) {
            /**底层的状态要提前判断，netty抛出的异常上层catch不到*/
            boolean isW = currentChannel.isWritable();
            boolean isC = currentChannel.isConnected();
            if (!(isW && isC)) {
                throw new RuntimeException("#sendRequest#channel is close!");
            }
            ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();
            buffer.writeBytes(requset.getBytes(StandardCharsets.UTF_8));
            currentChannel.write(buffer);
            Log.d(TAG, String.format("%s:%d -- packet#send ok", mHost, mPort));
            return true;
        } else {
            Log.e(TAG, String.format("%s:%d -- packet#send failed", mHost, mPort));
            return false;
        }
    }
}
